FortiGate VM の syslog を S3 に保管して Athena で分析してみた

こんにちは、大前です。

以前、EC2 で立てた FortiGate から syslog を出力することをやってみました。

今回は出力した syslog を CloudWatch Logs + Kinesis Firehose 経由で S3 に保存し、Athena を利用した分析までやってみました。

構成図

今回最終的に用意する構成は以下となります。

やってみた

（事前準備） FortiGate VM の構築 〜 syslog の出力 設定

以下のブログを参考に、FortiGate VM の構築と、FortiGate VM 上の syslog が syslog サーバーとして立てた EC2 に出力される状態にしておきます。

FortiGate VM 上でのログ出力設定については先人のブログが多数ありますので、こちらもご参照ください。

AWS 上の状態としては下記のイメージとなります。FortiGate VM の syslog が syslog サーバー（EC2）の /var/log/fortilog に出力されるようにしました。

CloudWatch Logs への syslog 出力

syslog サーバーに CloudWatch Agent をインストールし、FortiGate VM から出力される syslog を CloudWatch Logs に出力するようにします。

事前に以下の SSM パラメータを作成しておきました。FortiGate VM からの syslog 出力先を変えている場合は、file_path 部分を適宜修正してください。

{
	"agent": {
		"run_as_user": "root"
	},
	"logs": {
		"logs_collected": {
			"files": {
				"collect_list": [
					{
						"file_path": "/var/log/fortilog",
						"log_group_name": "fortigate-syslog",
						"log_stream_name": "{instance_id}",
						"retention_in_days": 7
					}
				]
			}
		}
	}
}

SSM Run Command で CloudWatch Agent をインストールします。<インスタンスID> には syslog サーバーのインスタンス ID を指定します。

aws ssm send-command \
    --document-name "AWS-ConfigureAWSPackage" \
    --instance-ids "<インスタンスID>" \
    --parameters '{"action":["Install"],"installationType":["Uninstall and reinstall"],"name":["AmazonCloudWatchAgent"],"version":["latest"]}'

続いて、事前に用意しておいた SSM パラメータで CloudWatch Agent を起動します。<インスタンスID> と <パラメータ名> 部分を置き換えて実行します。

aws ssm send-command \
  --document-name "AmazonCloudWatch-ManageAgent" \
  --instance-ids "<インスタンスID>" \
  --parameters '{"action":["configure"],"mode":["ec2"],"optionalConfigurationLocation":["<パラメータ名>"],"optionalConfigurationSource":["ssm"],"optionalOpenTelemetryCollectorConfigurationSource":["ssm"],"optionalRestart":["yes"]}'

下記のように、SSM パラメータで指定した通りに CloudWatch Logs に syslog が出力されました。

構成の状態としては以下のイメージとなります。

Kinesis Firehose の作成

続いて、CloudWatch Logs 上の syslog を S3 に転送するための Kinesis Firehose を作成します。

CloudWatch Logs → Kinesis Firehose → S3 という構成は以下ブログなどを参考に構築しました。

一点、CloudWatch Logs 上のログをそのまま Kinesis Firehose 経由で S3 に出力してしまうと、Athena で扱いにくいデータとなってしまうため、下記ブログのように、Kinesis Firehose の設定でフォーマット変換を設定して S3 上でもオリジナルのログ形式で保管されるようにしておきます。

CloudWatch Logs Group にサブスクリプションフィルタ作成

Kinesis Firehose が用意できたら、CloudWatch Logs Group にサブスクリプションフィルタを設定します。

ログ形式とフィルター設定については、以下で設定しました。必要な IAM ロールなどは、適宜作成して設定します。

• ログ形式 ... その他
• サブスクリプションフィルターのパターン ... " "

ここまで設定し、S3 に FortiGate VM からの syslog が保管されることが確認できました。

構成の状態としては以下のイメージとなります。

Athena テーブルの作成

最後に、S3 上のログファイルに対し、Athena で分析を行うためのテーブルを作成します。

今回は以下の DDL を実行してテーブルを作成しました。データベース名やテーブル名、LOCATION については各々の環境に合わせて置き換えてください。

CREATE EXTERNAL TABLE IF NOT EXISTS `<データベース名>`.`<作成するテーブル名>` (
  `log_timestamp` string,
  `instance_id` string,
  `date` string,
  `time` string,
  `devname` string,
  `devid` string,
  `eventtime` string,
  `tz` string,
  `logid` string,
  `type` string,
  `subtype` string,
  `level` string,
  `vd` string,
  `srcip` string,
  `srcport` string,
  `srcintf` string,
  `srcintfrole` string,
  `dstip` string,
  `dstport` string,
  `dstintf` string,
  `dstintfrole` string,
  `srccountry` string,
  `dstcountry` string,
  `sessionid` string,
  `proto` string,
  `action` string,
  `policyid` string,
  `policytype` string,
  `poluuid` string,
  `policyname` string,
  `service` string,
  `trandisp` string,
  `transip` string,
  `transport` string,
  `duration` string,
  `sentbyte` string,
  `rcvdbyte` string,
  `sentpkt` string,
  `rcvdpkt` string,
  `appcat` string
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
  'input.regex' = '(.{15}) ([^ ]*) date=([^ ]*) time=([^ ]*) devname="([^"]*)" devid="([^"]*)" eventtime=([^ ]*) tz=([^ ]*) logid="([^"]*)" type="([^"]*)" subtype="([^"]*)" level="([^"]*)" vd="([^"]*)" srcip=([^ ]*) srcport=([^ ]*) srcintf="([^"]*)" srcintfrole="([^"]*)" dstip=([^ ]*) dstport=([^ ]*) dstintf="([^"]*)" dstintfrole="([^"]*)" srccountry="([^"]*)" dstcountry="([^"]*)" sessionid=([^ ]*) proto=([^ ]*) action="([^"]*)" policyid=([^ ]*) policytype="([^"]*)" poluuid="([^"]*)" policyname="([^"]*)" service="([^"]*)" trandisp="([^"]*)" transip=([^ ]*) transport=([^ ]*) duration=([^ ]*) sentbyte=([^ ]*) rcvdbyte=([^ ]*) sentpkt=([^ ]*) rcvdpkt=([^ ]*) appcat="([^"]*)"'
)
STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://<バケット名>/<パス>/';

FortiGate VM から出力される syslog が date=XXXX time=XXXX といった風に、「カラム名=値」といった形式で出力されているため、input.regex の指定でうまく値だけ取り出すような記載にしています。

Athena で FortiGate VM の syslog を分析

Athena でクエリを実行する準備ができたので、試しに以下のクエリを実行してみます。subtype が forward であるログを取り出すためのクエリになります。

SELECT date, time, type, subtype, srcip, srcport, dstip, dstport  
    FROM "sampledb"."fortigate_syslog" 
    WHERE subtype = 'forward' 
    LIMIT 10;

下記のように、問題なくクエリ実行が成功し、FortiGate VM 上の syslog を Athena で分析することができました。

おわりに

EC2 上に立てた FortiGate VM の syslog を syslog サーバー → CloudWatch Logs → Kinesis Firehose → S3 の流れで出力し、S3 に保管されたログに対して Athena でクエリを実行するところまでやってみました。

あまり DDL は詳しく無いため、Athena のテーブル定義部分で苦労しましたが無事にクエリ実行までできたのでよかったです。（何か考慮不足あればご指摘ください） 今回はクエリ実行までできればよかったため考慮していませんが、実際の運用ではパーティション設定まで実施することが望ましいかと思います。

以上、AWS 事業本部の大前でした。

参考